package com.dtsheep.rxjava.stduy.observable;

import rx.Observable;
import rx.Subscription;
import rx.schedulers.Schedulers;

/**
 * SubscribeDemo
 *
 * <p>
 * </p>
 *
 * @author chuanbai
 * @date 2024/5/17
 */
public class SubscribeDemo {

    /**
     * 创建 Observable 被观察对象
     *
     * OnSubscribe 发射器 为函数式接口
     * Observable发射事件到Observer接收和处理事件，整个过程通常都是异步的。
     *
     * 它们之间的执行是串行的。也就是说，当第一次调用onNext("Hello")时，它会立即发出一个"Hello"事件；
     * 仅当第一个事件被完全处理后，才会发出"World"事件；
     * 最后，当所有事件（包括"Hello"和"World"）都被处理后，才会发出一个完成事件。
     *
     * 事件的处理与发射是异步的，因此无法保证事件的发射和处理顺序。
     * 具体来说，如果Observable和Observer之间有多个工作线程，那么事件的处理顺序可能不同于事件的发射顺序。
     *
     * @param
     * @return
     * @author chuanbai
     * @date ${DATE}
     */
    public void subscribeDemo01(){
        // 创建一个被观察对象
        Observable<String> observable = Observable.create(emitter -> {
            emitter.onNext("Hello");
            emitter.onNext("World");
            emitter.onCompleted();
        });

        // 订阅
        Subscription subscription = observable.subscribe(
                next -> System.out.println("Received: " + next),
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Complete")
        );

    }

    /**
     * 链式调用
     *
     * subscribeOn 是一个操作符，它允许你指定在哪个调度器上执行 Observable 的订阅逻辑。
     * subscribeOn 通常用于控制并发，确保 Observable 的订阅和数据发射在特定的线程或调度器上执行。
     *
     * subscribeOn 可以链式调用，因为它返回一个新的 Observable，这个
     * 新的 Observable 与原始的 Observable 行为相同，但是订阅逻辑会在指定的调度器上执行。
     *
     * @param
     * @return
     * @author chuanbai
     * @date ${DATE}
     */
    public void subscribeDemo02(){
        // 创建一个被观察对象  异步发射事件 cpu密集型
        Observable<Object> observable = Observable.create(emitter -> {
            emitter.onNext("Hello");
            emitter.onNext("World");
            emitter.onCompleted();
        }).subscribeOn(Schedulers.computation());

        // 观察者订阅
        Subscription subscription = observable.subscribe(
                next -> System.out.println("Received: " + next),
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Complete")
        );

        // 为了确保观察者能够接收到事件，可以手动控制程序的退出
        // 例如，可以使用一个无限循环或者一个长时间的等待
        try {
            Thread.sleep(5000); // 等待5秒，确保异步操作能够完成

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 如果被订阅 取消订阅
            if (!subscription.isUnsubscribed()) {
                subscription.unsubscribe(); // 取消订阅
            }
        }

    }

    public static void main(String[] args) {
        SubscribeDemo subscribeDemo = new SubscribeDemo();
        subscribeDemo.subscribeDemo02();
    }

}
